package com.gandong8.trace.tool.service.impl;

import com.gandong8.trace.tool.service.MqService;
import com.gandong8.trace.tool.util.JsonUtil;
import com.gandong8.trace.tool.util.TraceUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;


@Slf4j
@Service("kafkaService")
public class KafkaService implements MqService {
    @Autowired (required = false) @Qualifier( "kafkaTemplate" )
    //@Resource(name="kafkaTemplate")
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public Boolean send(String topic, Object object) {
        return send(topic,  JsonUtil.toStr(object));
    }

    @Override
    public Boolean send(String topic, String msg) {
        return send(topic, null, null, msg);
    }

    @Override
    public Boolean send(String topic, String tag, String key, Object object) {
        return send(topic, tag, key, JsonUtil.toStr(object));
    }

    @Override
    public Boolean send(String topic, String tag, String key, String msg) {
        try {
            log.info("开始发送消息：topic={},tag={},key={},msg={}" , topic, tag, key, msg);
            Message<String> message = MessageBuilder.withPayload(msg)
                    .setHeader(TraceUtil.TRACE_TAG, TraceUtil.getTraceId())
                    .setHeader("kafka_topic", topic)
                    .build();
            SendResult sendResult = kafkaTemplate.send(message).get();
            //SendResult sendResult = kafkaTemplate.send(topic ,key, msg).get();
            if(sendResult != null && sendResult.getRecordMetadata() != null) {
                log.info("发送成功");
                return true;
            }
            log.info("发送失败");
        } catch (Exception e) {
            log.error("发送异常", e);
            //throw new GlobalException("发送消息异常");
        }
        return false;
    }
}
